<?php
/**
 * Created by PhpStorm.
 * User: xiao
 * Date: 2018/8/30
 * Time: 10:54
 */

namespace app\index\controller;


use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;
use think\Controller;

class MessageTwo extends Controller
{

    public $normalQueue = 'cache_queue';
    public $normalExchange = 'cache_exchange';
    public $normalRouteKey = 'cache_route_key';


    public $deadQueue = 'dead_queue';
    public $deadExchange = 'dead_exchange';
    public $deadRouteKey = 'dead_route_key';


    public function client()
    {
        $connection = new AMQPStreamConnection('127.0.0.1', 5672, 'guest', 'guest');
        $channel = $connection->channel();

        // ########  死信的声明  ########
        $channel->exchange_declare($this->deadExchange, 'direct', false, true, false);
        $channel->queue_declare($this->deadQueue, false, true, false, false, false);
        $channel->queue_bind($this->deadQueue, $this->deadExchange, $this->deadRouteKey);
        // #############################

        $tale = new AMQPTable();
        $tale->set('x-dead-letter-exchange', $this->deadExchange);
        $tale->set('x-dead-letter-routing-key', $this->deadRouteKey);


        // ########  正常的声明  ########
        $channel->exchange_declare($this->normalExchange, 'direct', false, true, false);
        $channel->queue_declare($this->normalQueue, false, true, false, false, false, $tale);
        $channel->queue_bind($this->normalQueue, $this->normalExchange, $this->normalRouteKey);
        // #############################

        $msg = new AMQPMessage('Hello World!!!!!!!!!!Test' . date('Ymd H:i:s'), array(
            'expiration' => intval(10*1000),
            'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT
        ));

        // 先发到正常的队列中，当信息过期之后。会发到死信队列里面，服务端监听死信队列
        $channel->basic_publish($msg, $this->normalExchange, $this->normalRouteKey);
        echo date('Y-m-d H:i:s') . " [x] Sent 'Hello World!' " . PHP_EOL;

        $channel->close();
        $connection->close();
    }

    public function server()
    {
        $connection = new AMQPStreamConnection('127.0.0.1', 5672, 'guest', 'guest');
        $channel = $connection->channel();

        // ########  死信的声明  ########
        $channel->exchange_declare($this->deadExchange, 'direct', false, true, false);
        $channel->queue_declare($this->deadQueue, false, true, false, false, false);
        $channel->queue_bind($this->deadQueue, $this->deadExchange, $this->deadRouteKey);
        // #############################

//        // ########  正常的声明  ########
//        $channel->exchange_declare($this->normalExchange, 'direct', false, true, false);
//        // #############################

        echo ' [*] Waiting for message. To exit press CTRL+C ' . PHP_EOL;

        $callback = function ($msg) {
            echo date('Y-m-d H:i:s') . " [x] Received", $msg->body, PHP_EOL;
            $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
        };

        //只有consumer已经处理并确认了上一条message时queue才分派新的message给它
        $channel->basic_qos(null, 1, null);
        $channel->basic_consume($this->deadQueue, '', false, false, false, false, $callback);

        while (count($channel->callbacks)) {
            $channel->wait();
        }
        $channel->close();
        $connection->close();
    }


}